package com.example.window;

import com.example.model.WaterSensor;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.datastream.WindowedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.windows.GlobalWindow;

/**
 * Created with IntelliJ IDEA.
 * ClassName: KeyBy
 * Package: com.example.window
 * Description:
 * User: fzykd
 *
 * @Author: LQH
 * Date: 2023-07-23
 * Time: 13:13
 */

public class ProcessingKeyByWindow {
    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();


        SingleOutputStreamOperator<WaterSensor> map = env.socketTextStream("hadoop102", 7777)
                .map(new MapFunction<String, WaterSensor>() {
                    @Override
                    public WaterSensor map(String value) throws Exception {
                        final String[] s = value.split(" ");

                        return new WaterSensor(s[0], Long.valueOf(s[1]), Integer.valueOf(s[2]));
                    }
                });


        KeyedStream<WaterSensor, String> keyData = map.keyBy(value -> value.getId());

        //Key之后才能调用window方法
        //和聚合算子一样 KeyBy之后才能调用
        //参数也是一个窗口分配器
        WindowedStream<WaterSensor, String, GlobalWindow> window =
                //TumblingProcessingTimeWindows 滚动处理时间窗口 设定时间3秒 有第二个参数表示起始的偏移量
                //滚动窗口时间3秒 3秒之后 计算和关闭 处理时间窗口 改窗口进入第一条数据开始记时
                //keyData.window(TumblingProcessingTimeWindows.of(Time.seconds(3)));

                //滑动处理时间窗口 两个参数 窗口时常大小5秒 步长2
                ///keyData.window(SlidingProcessingTimeWindows.of(Time.seconds(5),Time.seconds(2)));

                //处理时间会话窗口 中间间隔有10秒没有数据来 就关闭窗口 并计算
                //keyData.window(ProcessingTimeSessionWindows.withGap(Time.seconds(10)));

                //在窗口课程中 没有细讲到 EventTime事件时间 因为还有每讲到时间语言
                //keyData.window(TumblingEventTimeWindows)

                //计数窗口
                //一个参数 表示计数的个数 就是滚动窗口 当有5条数据就会 触发计算 关闭窗口
                //两个参数 表示计数的个数 和 移动的步长 就是滑动窗口 每个窗口 统计5个数据当有2个数据就会统计输出结果
                keyData.countWindow(5,2);

        /**
         * 窗口操作主要有两部分 对KeyedStream调用window方法 参数是窗口分配器 windowAssigners
         * 指定窗口的类型 生成WindowedStream 在调用 窗口函数 是定义窗口中具体的处理逻辑
         */

        //窗口函数的规约函数
        SingleOutputStreamOperator<WaterSensor> reduce = window.reduce(new ReduceFunction<WaterSensor>() {
            @Override
            public WaterSensor reduce(WaterSensor value1, WaterSensor value2) throws Exception {
                return new WaterSensor(value1.getId(), Long.valueOf(value1.getTs()) + value2.getTs(),
                        Integer.valueOf(value2.getVc()) + value2.getVc());
            }
        });

        reduce.print();

        env.execute();
    }
}
